package org.raymond.iworks.study.basic.thread.pool;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.*;

import static java.util.concurrent.TimeUnit.MILLISECONDS;

/**
 * LinkedBlockingQueue用在newSingleThreadExecutor和newFixedThreadPool上,而newCachedThreadPool却用SynchronousQueue:
 * 1. 单线程池和固定线程池中,线程数量是有限的,因此提交的任务需要在LinkedBlockingQueue队列中等待空余的线程;
 * 2. 而缓存线程池中,线程数量几乎无限(上限为Integer.MAX_VALUE),因此提交的任务只需要在SynchronousQueue队列中同步移交给空余线程即可,
 * 所以有时也会说SynchronousQueue的吞吐量要高于LinkedBlockingQueue和ArrayBlockingQueue
 *
 * LinkedTransferQueue是ConcurrentLinkedQueue,SynchronousQueue(在公平模式下),无界的LinkedBlockingQueues等的超集;
 * 允许你混合使用阻塞队列的多种特性.
 * 所以,在合适的场景中,请尽量使用LinkedTransferQueue.
 *
 * 固定大小线程池适合用于执行长期任务,缓存线程池适合执行短期任务,定时线程池适合执行周期性任务,单线程池适合需要按顺序执行的任务.
 */
@Slf4j
public class Demo1 {
    /**
     * 线程池线程数设置:
     * CPU密集型(加密,计算hash等):最佳线程数设置为CPU核心数的1——2倍
     * 耗时I/O型(读写数据库,文件,网络读写等):最佳线程数一般会大于CPU核心数很多倍,以JVM监控显示繁忙情况为依据,保证线程空闲可以衔接上.
     * 参考Brain Goezt推荐的计算方法:线程数=CPU核心数×(1+平均等待时间/平均工作时间)
     */
    public static void main(String[] args) {
        //testCacheThreadPool();
        //testFixedThreadPool();
        //testScheduledThreadPool();
        //testSingleThreadPool();
        testWorkStealingPool();
    }

    /**
     * 创建一个可根据需要创建新线程的线程池,但是在以前构造的线程可用时将重用它们,并在需要时使用提供的ThreadFactory创建新线程
     * 1. 线程池中数量没有固定,可达到最大值(Interger.MAX_VALUE)
     * 2. 只有非核心线程
     * 3. 有超时机制,默认时长为60s,即超过60s的空闲线程就会被回收
     * 4. 当线程池中的线程都处于活动状态时,线程池会创建新的线程来处理新任务,否则就会利用空闲的线程来处理新任务.
     * 因此任何任务都会被立即执行
     * 5. 该线程池比较适合执行大量耗时较少的任务
     */
    public static void testCacheThreadPool(){
        //Executors.newCachedThreadPool();
        // 主要问题是线程数最大数是Integer.MAX_VALUE,可能会创建数量非常多的线程,甚至OOM
        ThreadPoolExecutor executor = new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                60L, TimeUnit.SECONDS, new SynchronousQueue<>());
        for(int i=0; i<10; i++){
            final int task = i;
            executor.execute(()->{
                try {
                    Thread.sleep(20);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                for(int j=0; j<10; j++){
                    log.info("正在进行第{}个任务,第{}次循环.", task, j);
                }
            });
        }
        log.info("所有任务在缓存线程池上执行");
        executor.shutdown();
    }

    /**
     * 创建一个可根据需要创建新线程的线程池,但是在以前构造的线程可用时将重用它们,并在需要时使用提供的ThreadFactory创建新线程
     * 1.线程数量固定
     * 2.只有核心线程
     * 3.没有超时机制,线程不会被回收
     * 4.当所有线程都处于活动状态时,新任务都会处于等待状态,直到有线程空闲出来
     */
    public static void testFixedThreadPool(){
        //Executors.newFixedThreadPool(5);
        // 主要问题是堆积的请求处理队列可能会耗费非常大的内存,甚至OOM.(笔者注：阻塞队列均采用LinkedBlockingQueue)
        ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 5,
                0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>());
        for(int i=0; i<10; i++){
            final int task = i;
            executor.execute(()->{
                try {
                    Thread.sleep(20);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                for(int j=0; j<10; j++){
                    log.info("正在进行第{}个任务,第{}次循环.", task, j);
                }
            });
        }
        log.info("所有任务在固定线程池上执行");
        executor.shutdown();
    }

    /**
     * 只有一个核心线程,它确保所有的任务都在同一个线程中按顺序执行.因此在这些任务之间不需要处理线程同步的问题
     */
    public static void testSingleThreadPool(){
        //Executors.newSingleThreadExecutor();
        ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 1,
                0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>());
        for(int i=0; i<3; i++){
            final int task = i;
            executor.execute(()->{
                try {
                    Thread.sleep(20);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                for(int j=0; j<3; j++){
                    log.info("正在进行第{}个任务,第{}次循环.", task, j);
                }
            });
        }
        log.info("所有任务在单线程池上执行");
        executor.shutdown();
    }

    /**
     * 创建一个线程池,它可安排在给定延迟后运行命令或者定期地执行
     * 1.核心线程数量是固定的,而非核心线程数不固定的,并且非核心线程有超时机制,只要处于闲置状态就会被立即回收
     * 2.该线程池主要用于执行定时任务和具有固定周期的重复任务
     */
    public static void testScheduledThreadPool(){
        //Executors.newScheduledThreadPool(2);
        ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(5);
        for(int i=0; i<2; i++){
            final int task = i;
            executor.schedule(()->{
                try {
                    Thread.sleep(20);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                for(int j=0; j<2; j++){
                    log.info("正在进行第{}个任务,第{}次循环.", task, j);
                }
            }, 2, TimeUnit.SECONDS);
        }
        log.info("所有任务在定时线程池上执行");
        executor.shutdown();
    }

    /**
     * 创建一个线程池,该线程池维护足以支持给定并行度级别的线程,并可以使用多个队列来减少争用.
     * 并行度级别对应于活动参与或可用于参与任务处理的最大线程数.
     * 实际的线程数可能会动态增长和收缩.
     * newWorkStealingPool不能保证提交任务的执行顺序
     *
     * 工作窃取核心思想是,自己的活干完了去看看别人有没有没干完的活,如果有就拿过来帮他干.
     * 大多数实现机制是:为每个工作者程分配一个双端队列(本地队列)用于存放需要执行的任务,
     * 当自己的队列没有数据的时候从其它工作者队列中获得一个任务继续执行
     */
    public static void testWorkStealingPool(){
        Executors.newWorkStealingPool();
        ForkJoinPool steal = new ForkJoinPool(Runtime.getRuntime().availableProcessors(),
                        ForkJoinPool.defaultForkJoinWorkerThreadFactory,
                        null, true);
        ThreadPoolExecutor cacheThread = new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                60L, TimeUnit.SECONDS, new SynchronousQueue<>());

        Runnable r1 = ()->{
            log.info("working on 1");
        };
        Runnable r2 = ()->{
            log.info("working on 2");
        };
        steal.submit(r1);
        steal.submit(r2);
        log.info("所有任务在工作窃取线程池上执行");
        steal.shutdown();
    }
}
